package course.java.simulation.bankandexcute;

import course.java.simulation.common.*;
import lombok.Data;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.math.BigInteger;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 银行线程
 */

public class Bank extends BankFacade implements Runnable {
    Logger logger = LoggerFactory.getLogger(Bank.class);

    // volatile关键字保证账户可见性
    volatile Map<String, Double> bankMap;
    AtomicBigInteger transactionId;

    LinkedBlockingQueue<RequestObject> blockingQueue;

    Map<String, Semaphore> locks;

    volatile Map<String, AtomicInteger> TransactionIdMap = new ConcurrentHashMap<>();

    public Bank(HashMap<String, Double> initialUserAmount) {
        super(initialUserAmount);
        if ((initialUserAmount == null)) throw new RuntimeException("");
        bankMap = new ConcurrentHashMap<>(initialUserAmount);
        transactionId = new AtomicBigInteger(BigInteger.ZERO);

        locks = new ConcurrentHashMap<>();

        for (String userId : initialUserAmount.keySet()) {
            TransactionIdMap.put(userId, new AtomicInteger(0));
        }
    }

    public LinkedBlockingQueue<RequestObject> getBlockingQueue() {
        return blockingQueue;
    }

    public void setBlockingQueue(LinkedBlockingQueue<RequestObject> blockingQueue) {
        this.blockingQueue = blockingQueue;
    }

    @Override
    public String studentID() {
        return "ngb113";
    }

    @Override
    public Optional<TransactionCommands> openTransaction(String userId) {
        if (bankMap.containsKey(userId)) {
            return Optional.of(new TransactionCommands() {
                boolean DoneState, AbortedState, CommittedState;
                // 取出或存入的钱，取出用-，存入用+
                double totalLocalOperations;
                List<Operation> successfulOperationList;
                List<Operation> ignoredOperationList;
                BigInteger currentTransactionId;
                // 当前账户未提交过默认为0
                AtomicInteger currentCmtVersion;

                {
                    successfulOperationList = new ArrayList<>();
                    ignoredOperationList = new ArrayList<>();
                    totalLocalOperations = 0;
                    DoneState = AbortedState = CommittedState = false;
                    currentTransactionId = transactionId.incrementAndGet();
                    currentCmtVersion = TransactionIdMap.get(userId);
                }

                @Override
                public BigInteger getTransactionId() {
                    return currentTransactionId;
                }

                @Override
                public double getTentativeTotalAmount() {
                    if (DoneState)
                        return bankMap.get(userId);
                    else
                        return -1;
                }

                @Override
                public boolean withdrawMoney(double amount) {
                    if ((amount < 0) || (DoneState)) {
                        ignoredOperationList.add(Operation.Withdraw(amount, ignoredOperationList.size()));
                        return false;
                    } else {
                        double val = bankMap.get(userId);
                        if (val >= amount) {
                            successfulOperationList.add(Operation.Withdraw(amount, successfulOperationList.size()));
                            totalLocalOperations -= amount;
                            return true;
                        } else {
                            ignoredOperationList.add(Operation.Withdraw(amount, ignoredOperationList.size()));
                            return false;
                        }
                    }
                }

                @Override
                public boolean payMoneyToAccount(double amount) {
                    if ((amount < 0) || (DoneState)) {
                        ignoredOperationList.add(Operation.Pay(amount, ignoredOperationList.size()));
                        return false;
                    }
                    else {
                        successfulOperationList.add(Operation.Pay(amount, successfulOperationList.size()));
                        totalLocalOperations += amount;
                        return true;
                    }
                }

                @Override
                public void abort() {
                    if (!DoneState) {
                        DoneState = AbortedState = true;
                        CommittedState = false;
                    }
                }

                @Override
                public CommitResult commit() {
                    //信号量机制为每个账户分配互斥锁
                    Semaphore semaphore = getLock(userId);
                    try {
                        semaphore.acquire();
                        // 判断是否有提交，有提交则判定此次无法提交
                        if (!DoneState && currentCmtVersion.equals(TransactionIdMap.get(userId))) {
                            AbortedState = false;
                            DoneState = CommittedState = true;
                            bankMap.computeIfPresent(userId, (s, aDouble) -> aDouble += totalLocalOperations);
                            // 版本号机制，提交之后++
                            TransactionIdMap.put(userId, new AtomicInteger(currentCmtVersion.incrementAndGet()));
                            return new CommitResult(successfulOperationList, ignoredOperationList, bankMap.get(userId));
                        } else {
                            return null;
                        }
                    } catch (InterruptedException e) {
                        logger.error("提交失败，失败账户为：{},异常信息为：{}", userId, e.getMessage());
                        return null;
                    } finally {
                        semaphore.release();
                    }
                }
            });
        } else
            return Optional.empty();
    }

    @Override
    public void run() {
        RequestObject requestObject = null;
        while (true) {
            try {
                requestObject =  blockingQueue.take();
                new Thread(new TradingExecute(this, requestObject)).start();
            } catch (InterruptedException e) {
                logger.error("出现异常，开启失败:{}", e.getMessage());
            }
        }
    }

    /**
     * 每个账户通过信号量获得互斥锁
     * @param userId
     * @return
     */
    private Semaphore getLock(String userId){
        if(locks.get(userId) == null){
            //double check
            synchronized (userId){
                if(locks.get(userId) == null){
                    Semaphore semaphore = new Semaphore(1);
                    locks.put(userId, semaphore);
                    return semaphore;
                }
            }
        }
        return locks.get(userId);
    }
}
